package org.demo.test.springbatch;

import com.hundsun.jrescloud.batch.mq.constant.Queues;
import com.hundsun.jrescloud.batch.runtime.springbatch.job.BatchJobRepository;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.batch.integration.partition.BeanFactoryStepLocator;
import org.springframework.batch.integration.partition.StepExecutionRequestHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.amqp.inbound.AmqpInboundGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.QueueChannel;

@Configuration
public class ClusterJobSlaveAutoConfig {
    @Autowired
    private ApplicationContext applicationContext;

    @Autowired
    @Qualifier("batchConnectionFactory")
    private ConnectionFactory connectionFactory;

    @Bean("slaveRequestListenerContainer")
    public SimpleMessageListenerContainer splitStepListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueueNames(Queues.JOB_SPLIT);
        container.setAutoStartup(false);
        return container;
    }

    @Bean
    public AmqpInboundGateway amqpInboundGateway(@Qualifier("slaveRequestListenerContainer") SimpleMessageListenerContainer container) {
        AmqpInboundGateway gateway = new AmqpInboundGateway(container);
        gateway.setReplyTimeout(-1);
        gateway.setRequestChannel(slaveRequestChannel());
        gateway.setReplyChannel(slaveReplyChannel());
        return gateway;
    }

    @Bean
    @ServiceActivator(inputChannel = "slaveRequestChannel", outputChannel = "slaveReplyChannel")
    public StepExecutionRequestHandler stepExecutionRequestHandler(BatchJobRepository repository) {
        StepExecutionRequestHandler stepExecutionRequestHandler = new StepExecutionRequestHandler();
        BeanFactoryStepLocator stepLocator = new BeanFactoryStepLocator();
        stepLocator.setBeanFactory(applicationContext);
        stepExecutionRequestHandler.setStepLocator(stepLocator);
        stepExecutionRequestHandler.setJobExplorer(repository.getJobExplorer());
        return stepExecutionRequestHandler;
    }

    @Bean("slaveRequestChannel")
    public QueueChannel slaveRequestChannel() {
        return new QueueChannel();
    }

    @Bean("slaveReplyChannel")
    public QueueChannel slaveReplyChannel() {
        return new QueueChannel();
    }

}
